package com.jz.dynadatasorces.rocketMQ.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class ConsumerService {
/*
    @Value("${rocketmq.name-server}")
    private String nameServer;

    private DefaultMQPullConsumer pullConsumer;

    public MyService() {
        this.pullConsumer = new DefaultMQPullConsumer("your-consumer-group");
        this.pullConsumer.setNamesrvAddr(nameServer);
        this.pullConsumer.start();
    }

    // 调用这个方法来消费消息
    public void consumeMessages() {
        try {
            // 指定Topic和队列ID
            String topic = "msg";
            int queueId = 0; // 根据实际情况选择队列ID

            // 拉取消息
            PullResult pullResult = pullConsumer.pullBlockIfNotFound(topic, queueId, 0L, null, 32);
            switch (pullResult.getPullStatus()) {
                case FOUND:
                    // 处理消息
                    List<MessageExt> msgFoundList = pullResult.getMsgFoundList();
                    for (MessageExt msg : msgFoundList) {
                        // 你的业务逻辑处理
                      System.out.println("Received message: " + new String(msg.getBody()));

                    }
                    // 提交offset
                    pullConsumer.updateConsumeOffset(topic, queueId, pullResult.getNextBeginOffset());
                    break;
                case NO_MATCHED_MSG:
                case NO_NEW_MSG:
                case OFFSET_ILLEGAL:
                default:
                    // 根据情况处理
                    break;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }*/
}

